package com.qdu;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;

import java.io.IOException;

public class HBaseScan {
    public static void main(String[] args) {
        // HBase的连接以及扫描t_log的一定Key范围的数据
        Configuration conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", "niit-master");
        conf.setInt("hbase.rpc.timeout", 20000);
        conf.setInt("hbase.client.operation.timeout", 20000);
        conf.setInt("hbase.client.scanner.timeout.period", 20000);

        // 连接HBase并扫描数据的逻辑可以在这里实现
        try (
                Connection conn = ConnectionFactory.createConnection(conf);
                Table tlog = conn.getTable(TableName.valueOf("t_log"))
        ) {

            // 新建一个Scan对象
            Scan scan = new Scan();
            // 设置扫描的起始RowKey和结束RowKey，对Scan命令添加过滤条件
            scan.withStartRow("a1".getBytes());
            scan.withStopRow("zz".getBytes());
             // 设置每次扫描返回的最大行数
            scan.setLimit(10);
            // 创建行过滤器，可用按照RowKey进行个性化过滤
            RowFilter rowFilter = new RowFilter(
                    CompareOperator.EQUAL,
                    new SubstringComparator("a")
            );
            // 将行过滤器加入scan对象中
            scan.setFilter(rowFilter);
            // 设置值过滤器
            ValueFilter valueFilter = new ValueFilter(
                    CompareOperator.EQUAL,
                    new SubstringComparator("3")
            );
            // 将值过滤器加入scan对象中
            scan.setFilter(valueFilter);
            // 执行扫描操作并处理结果
            ResultScanner resultScanner= tlog.getScanner(scan);

            // 使用双重循环，从scan结果中，逐行，逐列显示数据
            for (Result row:resultScanner){
                for (Cell cell : row.rawCells()){
                    System.out.print(
                            "列族=" + new String(CellUtil.cloneFamily(cell))
                    );
                    System.out.print(
                            "列标识符=" + new String(CellUtil.cloneQualifier(cell))
                    );
                    System.out.println(
                            "Value=" + new String(CellUtil.cloneValue(cell))
                    );

                }
                System.out.println("------------------");
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }
}
